package com.atguigu.flink.edu.app.dwd.db;

import com.atguigu.flink.edu.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class dwd_Trade_Pay_order_Suc {
    public static void main(String[] args) {
        //TODO 1. 基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表指定环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2. 检查点相关设置(略)
        //TODO 3.从topic_db主题中读取数据创建动态表 指定Watermark以及提取事件时间字段
        tableEnv.executeSql("create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`proc_time` as PROCTIME(),\n" +
                "`ts` string,\n" +
                "row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))),\n" +
                "watermark for row_time as row_time" +
                ")" + MyKafkaUtil.getKafakDDDL("topic_db", "dwd_trade_pay_order_suc"));

        //TODO 4.过滤出支付成功数据
        Table paymentInfo = tableEnv.sqlQuery("select\n" +
                        "data['order_id'] order_id,\n" +
                        "data['payment_type'] payment_type_code,\n" +
                        "data['callback_time'] callback_time,\n" +
                        "row_time,\n" +
                        "`proc_time`,\n" +
                        "ts\n" +
                        "from topic_db\n" +
                        "where `table` = 'payment_info'\n"
        );
        tableEnv.createTemporaryView("payment_info", paymentInfo);
        //tableEnv.executeSql("select * from payment_info").print();

        //TODO 5.从kafka的下单主题中读取下单数据
        Table orderDetail = tableEnv.sqlQuery("select \n" +
                "data['user_id'] user_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['course_id'] course_id,\n" +
                "data['course_name'] course_name,\n" +
                "data['create_time'] create_time,\n" +
                "data['origin_amount'] origin_amount,\n" +
                "data['final_amount'] final_amount\n" +
                "from `topic_db` where `table` = 'order_detail'\n" +
                "and `type` = 'insert'\n");
        tableEnv.createTemporaryView("order_detail", orderDetail);
        //tableEnv.executeSql("select * from order_detail").print();

        //TODO 7.关联3张表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "pi.order_id,\n" +
                "od.user_id,\n" +
                "od.course_id,\n" +
                "od.course_name,\n" +
                "pi.payment_type_code,\n" +
                "pi.callback_time,\n" +
                "od.origin_amount,\n" +
                "od.final_amount final_amount,\n" +
                "pi.ts \n" +
                "from payment_info pi left join order_detail od \n" +
                "on od.order_id = pi.order_id\n " );
        tableEnv.createTemporaryView("result_table", resultTable);
        //tableEnv.executeSql("select * from result_table").print();

        //TODO 8.将关联的结果写到kafka主题
        tableEnv.executeSql("create table dwd_trade_pay_order_suc(\n" +
                "order_id string,\n" +
                "user_id string,\n" +
                "course_id string,\n" +
                "course_name string,\n" +
                "payment_type_code string,\n" +
                "callback_time string,\n" +
                "origin_amount string,\n" +
                "final_amount string,\n" +
                "ts string,\n" +
                "primary key(order_id) not enforced\n" +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_pay_order_suc"));

        // TODO 9. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("insert into dwd_trade_pay_order_suc select * from result_table");
        tableEnv.executeSql("select * from result_table").print();
    }
}
